using System;
using System.Diagnostics;
using System.Collections.Generic;

//
// * Copyright (c) 2002-2009 "Neo Technology,"
// *     Network Engine for Objects in Lund AB [http://neotechnology.com]
// *
// * This file is part of Neo4j.
// * 
// * Neo4j is free software: you can redistribute it and/or modify
// * it under the terms of the GNU Affero General Public License as
// * published by the Free Software Foundation, either version 3 of the
// * License, or (at your option) any later version.
// * 
// * This program is distributed in the hope that it will be useful,
// * but WITHOUT ANY WARRANTY; without even the implied warranty of
// * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// * GNU Affero General Public License for more details.
// * 
// * You should have received a copy of the GNU Affero General Public License
// * along with this program. If not, see <http://www.gnu.org/licenses/>.
// 
namespace org.neo4j.kernel.impl.nioneo.store
{


///
/// <summary> * Manages <seealso cref="PersistenceWindow persistence windows"/> for a store. Each store
/// * can configure how much memory it has for
/// * <seealso cref="MappedPersistenceWindow memory mapped windows"/>. This class tries to
/// * make the most efficient use of those windows by allocating them in such a way
/// * that the most frequently used records/blocks (be it for read or write
/// * operations) are encapsulated by a memory mapped persistence window. </summary>
/// 
	internal class PersistenceWindowPool
	{
		private const int MAX_BRICK_COUNT = 10000;

		private readonly string storeName;
		private readonly int blockSize;
		private FileChannel fileChannel;
		private Map<int, PersistenceRow> activeRowWindows = new Dictionary<int, PersistenceRow>();
		private long availableMem = 0;
		private long memUsed = 0;
		private int brickCount = 0;
		private int brickSize = 0;
		private BrickElement[] brickArray = new BrickElement[0];
		private int brickMiss = 0;

		private static Logger log = Logger.getLogger(typeof(PersistenceWindowPool).Name);
		private const int REFRESH_BRICK_COUNT = 5000;
		private readonly FileChannel.MapMode mapMode;

		private int hit = 0;
		private int miss = 0;
		private int switches = 0;
		private int ooe = 0;
		private bool useMemoryMapped = true;

///    
///     <summary> * Create new pool for a store.
///     *  </summary>
///     * <param name="storeName">
///     *            Name of store that use this pool </param>
///     * <param name="blockSize">
///     *            The size of each record/block in the store </param>
///     * <param name="fileChannel">
///     *            A fileChannel to the store </param>
///     * <param name="mappedMem">
///     *            Number of bytes dedicated to memory mapped windows </param>
///     * <exception cref="IOException">
///     *             If unable to create pool </exception>
///     
		internal PersistenceWindowPool(string storeName, int blockSize, FileChannel fileChannel, long mappedMem, bool useMemoryMappedBuffers, bool readOnly)
		{
			this.storeName = storeName;
			this.blockSize = blockSize;
			this.fileChannel = fileChannel;
			this.availableMem = mappedMem;
			this.useMemoryMapped = useMemoryMappedBuffers;
			if (readOnly)
			{
				mapMode = FileChannel.MapMode.READ_ONLY;
			}
			else
			{
				mapMode = FileChannel.MapMode.READ_WRITE;
			}
			setupBricks();
			dumpStatus();
		}

///    
///     <summary> * Acquires a windows for <CODE>position</CODE> and <CODE>operationType</CODE>
///     * locking the window preventing other threads from using it.
///     *  </summary>
///     * <param name="position">
///     *            The position the needs to be encapsulated by the window </param>
///     * <param name="operationType">
///     *            The type of operation (READ or WRITE) </param>
///     * <returns> A locked window encapsulating the position </returns>
///     * <exception cref="IOException">
///     *             If unable to acquire the window </exception>
///     
		internal virtual PersistenceWindow acquire(long position, OperationType operationType)
		{
			LockableWindow window = null;
			lock (activeRowWindows)
			{
				if (brickMiss >= REFRESH_BRICK_COUNT)
				{
					brickMiss = 0;
					refreshBricks();
				}
				if (brickSize > 0)
				{
					int brickIndex = (int)(position * blockSize / brickSize);
					if (brickIndex < brickArray.Length)
					{
						window = brickArray[brickIndex].getWindow();
					// assert window == null || window.encapsulates( position );
						brickArray[brickIndex].setHit();
					}
					else
					{
						expandBricks(brickIndex + 1);
						window = brickArray[brickIndex].getWindow();
					}
				}
				if (window == null)
				{
					miss++;
					brickMiss++;

					PersistenceRow dpw = activeRowWindows.get((int) position);

					if (dpw == null)
					{
						dpw = new PersistenceRow(position, blockSize, fileChannel);
						if (operationType == OperationType.READ)
						{
							dpw.readPosition();
						}
					}
					window = dpw;
					activeRowWindows.put((int) position, dpw);
				}
				else
				{
					hit++;
				}
				window.mark();
			}
			window.lock();
			window.setOperationType(operationType);
			return window;
		}

		internal virtual void dumpStatistics()
		{
			log.finest(storeName + " hit=" + hit + " miss=" + miss + " switches=" + switches + " ooe=" + ooe);
		}

///    
///     <summary> * Releases a window used for an operation back to the pool and unlocks it
///     * so other threads may use it.
///     *  </summary>
///     * <param name="window">
///     *            The window to be released </param>
///     * <exception cref="IOException">
///     *             If unable to release window </exception>
///     
		internal virtual void release(PersistenceWindow window)
		{
			lock (activeRowWindows)
			{
				if (window is PersistenceRow)
				{
					PersistenceRow dpw = (PersistenceRow) window;
				// will only write if operation was write
					dpw.writeOut();
					if (dpw.getWaitingThreadsCount() == 0 && !dpw.isMarked())
					{
						int key = (int) dpw.position();
						activeRowWindows.Remove(key);
					}
				}
				((LockableWindow) window).unLock();
			}
		}

		internal virtual void close()
		{
			flushAll();
			lock (activeRowWindows)
			{
				foreach (BrickElement element in brickArray)
				{
					if (element.getWindow() != null)
					{
						element.getWindow().Close();
						element.setWindow(null);
					}
				}
				fileChannel = null;
				activeRowWindows.Clear();
			}
			activeRowWindows = null;
			dumpStatistics();
		}

		internal virtual void flushAll()
		{
			lock (activeRowWindows)
			{
				foreach (BrickElement element in brickArray)
				{
					if (element.getWindow() != null)
					{
						element.getWindow().force();
					}
				}
			}
			try
			{
				fileChannel.force(false);
			}
			catch (IOException e)
			{
				throw new UnderlyingStorageException("Failed to flush file channel " + storeName, e);
			}
		}

		private class BrickElement
		{
			private int index;
			private int hitCount;
			private LockableWindow window = null;

			internal BrickElement(int index)
			{
				this.index = index;
			}

			internal virtual void setWindow(LockableWindow window)
			{
				this.window = window;
			}

			internal virtual LockableWindow getWindow()
			{
				return window;
			}

			internal virtual int index()
			{
				return index;
			}

			internal virtual void setHit()
			{
				hitCount += 10;
				if (hitCount < 0)
				{
					hitCount -= 10;
				}
			}

			internal virtual int getHit()
			{
				return hitCount;
			}

			internal virtual void refresh()
			{
				if (window == null)
				{
					hitCount /= 1.25;
				}
				else
				{
					hitCount /= 1.15;
				}
			}

			public override string ToString()
			{
				return "" + hitCount + (window == null ? "x" : "o");
			}
		}

		private void setupBricks()
		{
			long fileSize = -1;
			try
			{
				fileSize = fileChannel.size();
			}
			catch (IOException e)
			{
				throw new UnderlyingStorageException("Unable to get file size for " + storeName, e);
			}
			if (blockSize == 0)
			{
				return;
			}
			if (availableMem > 0 && availableMem < blockSize * 10l)
			{
				logWarn("Unable to use " + availableMem + "b as memory mapped windows, need at least " + blockSize * 10 + "b (block size * 10)");
				logWarn("Memory mapped windows have been turned off");
				availableMem = 0;
				brickCount = 0;
				brickSize = 0;
				return;
			}
			if (availableMem > 0 && fileSize > 0)
			{
				double ratio = (availableMem + 0.0d) / fileSize;
				if (ratio >= 1)
				{
					brickSize = (int)(availableMem / 100);
					if (brickSize < 0)
					{
						brickSize = int.MaxValue;
					}
					brickSize = (brickSize / blockSize) * blockSize;
					brickCount = (int)(fileSize / brickSize);
				}
				else
				{
					brickCount = (int)(100.0d / ratio);
					if (brickCount > MAX_BRICK_COUNT)
					{
						brickCount = MAX_BRICK_COUNT;
					}
					if (fileSize / brickCount > availableMem)
					{
						logWarn("Unable to use " + (availableMem / 1024) + "kb as memory mapped windows, need at least " + (fileSize / brickCount / 1024) + "kb");
						logWarn("Memory mapped windows have been turned off");
						availableMem = 0;
						brickCount = 0;
						brickSize = 0;
						return;
					}
					brickSize = (int)(fileSize / brickCount);
					if (brickSize < 0)
					{
						brickSize = int.MaxValue;
						brickSize = (brickSize / blockSize) * blockSize;
						brickCount = (int)(fileSize / brickSize);
					}
					else
					{
						brickSize = (brickSize / blockSize) * blockSize;
					}
					Debug.Assert(brickSize > blockSize);
				}
			}
			else if (availableMem > 0)
			{
				brickSize = (int)(availableMem / 100);
				if (brickSize < 0)
				{
					brickSize = int.MaxValue;
				}
				brickSize = (brickSize / blockSize) * blockSize;
			}
			brickArray = new BrickElement[brickCount];
			for (int i = 0; i < brickCount; i++)
			{
				BrickElement element = new BrickElement(i);
				brickArray[i] = element;
			}
		}

		private void refreshBricks()
		{
			if (brickSize <= 0)
			{
			// memory mapped turned off
				return;
			}
			List<BrickElement> nonMappedBricks = new List<BrickElement>();
			List<BrickElement> mappedBricks = new List<BrickElement>();
			for (int i = 0; i < brickCount; i++)
			{
				BrickElement be = brickArray[i];
				if (be.getWindow() != null)
				{
					mappedBricks.Add(be);
				}
				else
				{
					nonMappedBricks.Add(be);
				}
				be.refresh();
			}
			Collections.sort(nonMappedBricks, new BrickSorter());
			Collections.sort(mappedBricks, new BrickSorter());
			int mappedIndex = 0;
			int nonMappedIndex = nonMappedBricks.Count - 1;
		// fill up unused memory
			while (memUsed + brickSize <= availableMem && nonMappedIndex >= 0)
			{
				BrickElement nonMappedBrick = nonMappedBricks[nonMappedIndex--];
				if (nonMappedBrick.getHit() == 0)
				{
					return;
				}
				try
				{
					nonMappedBrick.setWindow(allocateNewWindow(nonMappedBrick.index()));
					memUsed += brickSize;
				}
				catch (MappedMemException e)
				{
					e.printStackTrace();
					ooe++;
					logWarn("Unable to memory map");
				}
				catch (OutOfMemoryError e)
				{
					e.printStackTrace();
					ooe++;
					logWarn("Unable to allocate direct buffer");
				}
			}

		// switch bad mappings
			while (nonMappedIndex >= 0 && mappedIndex < mappedBricks.Count)
			{
				BrickElement mappedBrick = mappedBricks[mappedIndex++];
				BrickElement nonMappedBrick = nonMappedBricks[nonMappedIndex--];
				if (mappedBrick.getHit() >= nonMappedBrick.getHit())
				{
					break;
				}
				LockableWindow window = mappedBrick.getWindow();
				if (window.getWaitingThreadsCount() == 0 && !window.isMarked())
				{
					if (window is MappedPersistenceWindow)
					{
						((MappedPersistenceWindow) window).unmap();
					}
					else if (window is PlainPersistenceWindow)
					{
						((PlainPersistenceWindow) window).writeOut();
					}
					mappedBrick.setWindow(null);
					memUsed -= brickSize;
					try
					{
						nonMappedBrick.setWindow(allocateNewWindow(nonMappedBrick.index()));
						memUsed += brickSize;
						switches++;
					}
					catch (MappedMemException e)
					{
						ooe++;
						logWarn("Unable to memory map");
					}
					catch (OutOfMemoryError e)
					{
						ooe++;
						logWarn("Unable to allocate direct buffer");
					}
				}
			}
		}

		private void expandBricks(int newBrickCount)
		{
			if (newBrickCount > brickCount)
			{
				BrickElement[] tmpArray = new BrickElement[newBrickCount];
				System.Array.Copy(brickArray, 0, tmpArray, 0, brickArray.Length);
				for (int i = brickArray.Length; i < tmpArray.Length; i++)
				{
					BrickElement be = new BrickElement(i);
					tmpArray[i] = be;
					if (memUsed + brickSize <= availableMem)
					{
						try
						{
							be.setWindow(allocateNewWindow(i));
							memUsed += brickSize;
						}
						catch (MappedMemException e)
						{
							ooe++;
							logWarn("Unable to memory map");
						}
						catch (OutOfMemoryError e)
						{
							ooe++;
							logWarn("Unable to allocate direct buffer");
						}
					}
				}
				brickArray = tmpArray;
				brickCount = tmpArray.Length;
			}
		}

		private LockableWindow allocateNewWindow(long brick)
		{
			if (useMemoryMapped)
			{
				 return new MappedPersistenceWindow(brick * brickSize / blockSize, blockSize, brickSize, fileChannel, mapMode);
			}
			PlainPersistenceWindow dpw = new PlainPersistenceWindow(brick * brickSize / blockSize, blockSize, brickSize, fileChannel);
			dpw.readPosition();
			return dpw;
		}
		[Serializable]
		internal class BrickSorter : Comparator<BrickElement>
		{
			public virtual int compare(BrickElement o1, BrickElement o2)
			{
				return o1.getHit() - o2.getHit();
			}

			public override bool Equals(object o)
			{
				if (o is BrickSorter)
				{
					return true;
				}
				return false;
			}

			public override int GetHashCode()
			{
				return 7371;
			}
		}

		private void dumpStatus()
		{
			try
			{
				log.fine("[" + storeName + "] brickCount=" + brickCount + " brickSize=" + brickSize + "b mappedMem=" + availableMem + "b (storeSize=" + fileChannel.size() + "b)");
			}
			catch (IOException e)
			{
				throw new UnderlyingStorageException("Unable to get file size for " + storeName, e);
			}
		}

		private void logWarn(string logMessage)
		{
			log.warning("[" + storeName + "] " + logMessage);
		}
	}
}